package com.sc.server.system.batch.conf;


import com.sc.server.system.auth.entity.Menu;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.List;

@Configuration
@EnableBatchProcessing
@Slf4j
public class RabbitMqBatchConfigiguration {
    Integer count = 0;
    @Autowired
    private JobListener jobListener;
    @Resource
    private JobBuilderFactory jobBuilderFactory;    //用于构建JOB
    @Resource
    private StepBuilderFactory stepBuilderFactory;  //用于构建Step
    private final Object lock = new Object();
    @Autowired
    private RabbitTemplate amqpTemplate;
    @Bean
    public Job processMqJob() {
        return jobBuilderFactory.get("mqJob")
                .incrementer(new RunIdIncrementer()).listener(jobListener)
                .flow(mqStep()).end().build();
    }
    @Bean
    public Step mqStep() {
        return  stepBuilderFactory.get("mqStep").<Menu, Menu>chunk(100).
                //捕捉到异常就重试,重试100次还是异常,JOB就停止并标志失败
                        faultTolerant().retryLimit(3).retry(Exception.class).skipLimit(100).skip(Exception.class).
                        reader(getMqReader()).
                //指定ItemProcessor
                        processor(getStrProcess()).
                //指定ItemWriter
                        writer(getStrWriter()).
                        build();
    }

    @Bean
    public ItemReader<? extends Menu> getMqReader() {
        return new ItemReader<Menu>() {
            @Override
            public Menu read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                Menu menu;
                //加同步锁，以防万一
                synchronized (lock) {
                    amqpTemplate.setMessageConverter(jsonMessageConverter());
                    menu = (Menu) amqpTemplate.receiveAndConvert("fanout.c");
                }
                return menu;
            }
        };
      }
    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
    public ItemProcessor<Menu,Menu> getStrProcess() {
        return new ItemProcessor<Menu, Menu>() {
            @Override
            public Menu process(Menu s) throws Exception {
                return s;
            }
        };
    }
    public  ItemWriter<Menu> getStrWriter() {
        return new ItemWriter<Menu>() {
            @Override
            public void write(List<? extends Menu> list) throws Exception {
                for (Menu menu : list) {
                    System.out.println(menu.toString());
                }
            }
        };
    }
}
